package com.shengzai.flink.tf

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo2FlatMap {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.readTextFile("data/words.txt")

    //scala api
    val scalaDS: DataStream[String] = linesDS.flatMap(line => line.split(","))
    //scalaDS.print()

    // java api
    val javaDS: DataStream[String] = linesDS.flatMap(new FlatMapFunction[String, String] {

      /**
       * flatMap: 每一条数据执行一次, 返回多个结果
       *
       * @param line      : 一行数据
       * @param collector : 用于将数据发送到下游,可以发送多次
       */
      override def flatMap(line: String, collector: Collector[String]): Unit = {
        val split: Array[String] = line.split(",")
        //循环将数据发送到下雨
        for (word <- split) {
          //发送数据到下游
          collector.collect(word)
        }

        //可以发送而外的数据到下游
        collector.collect("数加")
      }
    })

    javaDS.print()

    env.execute()
  }
}
